﻿#include "zmqClient.h"
#include "common.h"
/**
 * 创建
 * @param context 上下文
 * @param url 链接地址
 * @param identity 客户端标识
 */
static zmq::socket_t *zmq_client_init(zmq::context_t &context, std::string &url, std::string &identity)
{
    //ZMQ版本检测
    s_version_assert(4, 0);

    zmq::socket_t *worker = new zmq::socket_t(context, ZMQ_DEALER);

    // 为客户端设置一个标识符
    worker->setsockopt(ZMQ_IDENTITY, identity.c_str(), identity.length());
    worker->connect(url);

    // 不等待
    int linger = 0;
    worker->setsockopt(ZMQ_LINGER, &linger, sizeof(linger));

    //告知队列客户端已就绪
    std::cout << "I: client ready" << std::endl;
    s_send(*worker, "READY");

    return worker;
}

/**
 * 队列客户端 会一直保持连接 服务器断开会一直重试
 * @param url 服务器地址
 * @param identity 客户端标识符（唯一）
 * @param msg 消息返回函数
 */
int quence_client(std::string &url, std::string &identity, SimpleMsgCallBack msgCallBack)
{
    zmq::context_t context(1);
    zmq::socket_t *worker = zmq_client_init(context, url, identity);
    //  活动性判断 如果活动性为零 则认为服务器已断开连接。
    size_t liveness = HEARTBEAT_LIVENESS;
    size_t interval = INTERVAL_INIT;

    //  定期发送心跳
    int64_t heartbeat_at = s_clock() + HEARTBEAT_INTERVAL;

    while (1)
    {
        zmq::pollitem_t items[] = {{*worker, 0, ZMQ_POLLIN, 0}};
        zmq::poll(items, 1, HEARTBEAT_INTERVAL);

        if (items[0].revents & ZMQ_POLLIN)
        {
            zmsg msg(*worker);

            if (msg.parts() == 3)
            {
                liveness = HEARTBEAT_LIVENESS;
                if (msgCallBack)
                    msgCallBack(worker, msg, identity, msg.body()); //  消息处理函数
            }
            else
            {
                if (msg.parts() == 1 && strcmp(msg.body(), "HEARTBEAT") == 0)
                {
                    liveness = HEARTBEAT_LIVENESS;
                }
                else
                {
                    std::cout << "E:  invalid message" << std::endl;
                    msg.dump();
                }
            }
            interval = INTERVAL_INIT;
        }
        else if (--liveness == 0)
        {
            std::cout << "W:  heartbeat failure, can't reach queue" << std::endl;
            std::cout << "W:  reconnecting in " << interval << " msec…" << std::endl;
            s_sleep(interval);

            if (interval < INTERVAL_MAX)
            {
                interval *= 2;
            }
            delete worker;
            worker = zmq_client_init(context, url, identity);
            liveness = HEARTBEAT_LIVENESS;
        }

        // 到点发出心跳检测
        if (s_clock() > heartbeat_at)
        {
            heartbeat_at = s_clock() + HEARTBEAT_INTERVAL;
            std::cout << "I: worker heartbeat" << std::endl;
            s_send(*worker, "HEARTBEAT");
        }
    }
    delete worker;
    return 0;
}

/**
 * 基础连接客户端 会多次重连 失败后会返回-1 发送成功之后返回0
 * @param url 服务器地址
 * @param identity 客户端标识符（唯一）
 * @param msg 消息返回函数
 */
int simple_client(std::string &url, std::string &identity, SimpleMsgCallBack msgCallBack, std::string &message)
{
    zmq::context_t context(1);
    zmq::socket_t *worker = zmq_client_init(context, url, identity);
    int retries_left = REQUEST_RETRIES;
    bool isSend = false;
    while (1)
    {
        zmq::pollitem_t items[] = {{*worker, 0, ZMQ_POLLIN, 0}};
        zmq::poll(items, 1, HEARTBEAT_INTERVAL);

        if (items[0].revents & ZMQ_POLLIN)
        {
            zmsg msg(*worker);
            if (msg.parts() == 1 && strcmp(msg.body(), "HEARTBEAT") == 0)
            {
                zmsg msg(message.c_str());
                msg.wrap(identity.c_str(), NULL);
                msg.send(*worker);
//                break;
            }
            else
            {
                if (msgCallBack)
                    msgCallBack(worker, msg, identity, msg.body()); //  消息处理函数
                msg.dump();
                break;
            }
        }
        else if (--retries_left == 0)
        {
            std::cout << "E: server seems to be offline, abandoning" << std::endl;
            return -1;
        }
        else
        {
            Sleep(REQUEST_TIMEOUT);
            std::cout << "I: worker heartbeat" << std::endl;
            s_send(*worker, "HEARTBEAT");
        }
    }
    delete worker;
    return 0;
}

//在队列末尾插入工作线程，重置到期
//工作进程必须不在队列中
static void s_worker_append(std::vector<worker_t> &queue, std::string &identity, QueueAppendCallBack cacb)
{
    bool found = false;
    for (std::vector<worker_t>::iterator it = queue.begin(); it < queue.end(); it++)
    {
        //循环判断
        if (it->identity.compare(identity) == 0)
        {
            std::cout << "client already exit :" << identity.c_str() << std::endl;
            found = true;
            break;
        }
    }
    if (!found)
    {
        worker_t worker;
        worker.identity = identity;
        worker.expiry = s_clock() + HEARTBEAT_INTERVAL * HEARTBEAT_LIVENESS;
        queue.push_back(worker);
        if (cacb)
            cacb(queue, identity);
    }
}

//  从队列中删除（如果存在）
static void s_worker_delete(std::vector<worker_t> &queue, std::string &identity)
{
    for (std::vector<worker_t>::iterator it = queue.begin(); it < queue.end(); it++)
    {
        if (it->identity.compare(identity) == 0)
        {
            it = queue.erase(it);
            break;
        }
    }
}

//重置,必须存在
static void s_worker_refresh(std::vector<worker_t> &queue, std::string &identity)
{
    bool found = false;
    for (std::vector<worker_t>::iterator it = queue.begin(); it < queue.end(); it++)
    {
        if (it->identity.compare(identity) == 0)
        {
            it->expiry = s_clock() + HEARTBEAT_INTERVAL * HEARTBEAT_LIVENESS;
            found = true;
            break;
        }
    }
    if (!found)
    {
        std::cout << "client：" << identity << "not ready" << std::endl;
    }
}

//   寻找并杀死过期客户端
static void s_queue_purge(std::vector<worker_t> &queue, QueueQuitCallBack cqcb)
{
    int64_t clock = s_clock();
    for (std::vector<worker_t>::iterator it = queue.begin(); it < queue.end(); it++)
    {
        if (clock > it->expiry)
        {
            std::string identity = it->identity;
            it = queue.erase(it) - 1;
            if (cqcb)
                cqcb(queue, identity);
        }
    }
}

/**
 * 队列模式 可对多个客户端进行心跳检测
 * @param queue 客户端队列
 * @param port 服务器监测端口
 * @param msgCallBack 服务器端消息处理函数
 * @param cacb 客户端进入回调
 * @param cqcb 客户端退出回调
 */
int quence_service(int port, QueueMsgCallBack msgCallBack, QueueAppendCallBack cacb, QueueQuitCallBack cqcb)
{
    //zmq 版本检测
    s_version_assert(4, 0);

    //生成一个队列
    std::vector<worker_t> queue;

    //  准备上下文和套接字
    zmq::context_t context(1);
    zmq::socket_t backend(context, ZMQ_ROUTER);

    //服务器监控地址
    char url[20];
    sprintf(url, "%s%d", "tcp://*:", port);
    backend.bind(url);

    //  定期发送心跳
    int64_t heartbeat_at = s_clock() + HEARTBEAT_INTERVAL;

    while (1)
    {
        zmq::pollitem_t items[] = {{backend, 0, ZMQ_POLLIN, 0}};
        zmq::poll(items, 1, HEARTBEAT_INTERVAL);

        //  处理客户端上的活动
        if (items[0].revents & ZMQ_POLLIN)
        {
            zmsg msg(backend);
            std::string identity(msg.unwrap());

            //  如果不是控制消息，则将回复返回到客户端
            if (msg.parts() == 1)
            {
                if (strcmp(msg.address(), "READY") == 0)
                {
                    s_worker_delete(queue, identity);
                    s_worker_append(queue, identity, cacb);
                }
                else
                {
                    if (strcmp(msg.address(), "HEARTBEAT") == 0)
                    {
                        s_worker_refresh(queue, identity);
                    }
                    else
                    {
                        msg.dump();
                    }
                }
            }
            else
            {
                s_worker_append(queue, identity, cacb);
                if (msgCallBack)
                    msgCallBack(&backend, msg, queue, identity, msg.body()); //  消息处理函数
            }
        }

        //  如果时间到了 发送心跳。
        if (s_clock() > heartbeat_at)
        {
            for (std::vector<worker_t>::iterator it = queue.begin(); it < queue.end(); it++)
            {
                zmsg msg("HEARTBEAT");
                msg.wrap(it->identity.c_str(), NULL);
                msg.send(backend);
            }
            heartbeat_at = s_clock() + HEARTBEAT_INTERVAL;
        }
        s_queue_purge(queue, cqcb);
    }
    //  我们从不退出主循环 但无论如何，假装做正确的关机（作者原话）
    queue.clear();
    return 0;
}

/**
 * 点对点对指定客户端进行心跳检测
 * @param queue 客户端队列
 * @param port 服务器监测端口
 * @param msgCallBack 服务器端消息处理函数
 * @param cacb 客户端进入回调
 * @param cqcb 客户端退出回调
 */
int point_service(int port, string &clientId, SimpleMsgCallBack msgCallBack, SimpleAppendCallBack cacb, SimpleQuitCallBack cqcb)
{
    //  zmq 版本检测
    s_version_assert(4, 0);

    //  准备上下文和套接字
    zmq::context_t context(1);
    zmq::socket_t backend(context, ZMQ_ROUTER);

    worker_t client;

    //服务器监控地址
    char url[20];
    sprintf(url, "%s%d", "tcp://*:", port);
    backend.bind(url);

    //  定期发送心跳
    int64_t heartbeat_at = s_clock() + HEARTBEAT_INTERVAL;

    bool clientExit = false;

    while (1)
    {
        zmq::pollitem_t items[] = {{backend, 0, ZMQ_POLLIN, 0}};
        zmq::poll(items, 1, HEARTBEAT_INTERVAL);

        //  处理客户端上的活动
        if (items[0].revents & ZMQ_POLLIN)
        {
            zmsg msg(backend);
            std::string identity(msg.unwrap());

            //如果不是指定客户端的请求 忽视
            if (strcmp(identity.c_str(), clientId.c_str()) == 0)
            {
                //  如果不是控制消息，则将回复返回到客户端
                if (msg.parts() == 1)
                {
                    if (strcmp(msg.address(), "READY") == 0)
                    {
                        client.identity = identity;
                        client.expiry = s_clock() + HEARTBEAT_INTERVAL * HEARTBEAT_LIVENESS;
                        clientExit = true;
                        if (cacb)
                            cacb(identity);
                    }
                    else
                    {
                        if (strcmp(msg.address(), "HEARTBEAT") == 0)
                        {
                            client.expiry = s_clock() + HEARTBEAT_INTERVAL * HEARTBEAT_LIVENESS;
                        }
                        else
                        {
                            msg.dump();
                        }
                    }
                }
                else
                {
                    if (msgCallBack)
                        msgCallBack(&backend, msg, identity, msg.body()); //  消息处理函数
                }
            }
        }

        //当客户端确认连接时再进行心跳检测
        if (clientExit)
        {
            int64_t clock = s_clock();
            //  如果时间到了 发送心跳。
            if (clock > heartbeat_at)
            {
                zmsg msg("HEARTBEAT");
                msg.wrap(clientId.c_str(), NULL);
                msg.send(backend);
                heartbeat_at = clock + HEARTBEAT_INTERVAL;
            }
            if (clock > client.expiry)
            {
                clientExit = false;
                if (cqcb)
                    cqcb(clientId);
            }
        }
    }
    return 0;
}

/**
 * 基础模式 无心跳检测 只是循环接收消息
 * @param port 监听端口
 * @param simpleMsgCallBack 消息接收回调
 */
int simple_service(int port, SimpleMsgCallBack msgCallBack)
{
    //  zmq 版本检测
    s_version_assert(4, 0);

    //  准备上下文和套接字
    zmq::context_t context(1);
    zmq::socket_t backend(context, ZMQ_ROUTER);

    //服务器监控地址
    char url[20];
    sprintf(url, "%s%d", "tcp://*:", port);
    backend.bind(url);

    while (1)
    {
        zmq::pollitem_t items[] = {{backend, 0, ZMQ_POLLIN, 0}};
        zmq::poll(items, 1, HEARTBEAT_INTERVAL);

        //  处理客户端上的活动
        if (items[0].revents & ZMQ_POLLIN)
        {
            zmsg msg(backend);
            std::string identity(msg.unwrap());

            //  如果不是控制消息，则将回复返回到客户端
            if (msg.parts() == 1)
            {
                if (strcmp(msg.address(), "READY") == 0)
                {
                    zmsg msg("HEARTBEAT");
                    msg.wrap(identity.c_str(), NULL);
                    msg.send(backend);
                }
                else
                {
                    if (strcmp(msg.address(), "HEARTBEAT") == 0)
                    {
                        zmsg msg("HEARTBEAT");
                        msg.wrap(identity.c_str(), NULL);
                        msg.send(backend);
                    }
                    else
                    {
                        msg.dump();
                    }
                }
            }
            else
            {
                if (msgCallBack)
                    msgCallBack(&backend, msg, identity, msg.body()); //  消息处理函数
            }
        }
    }
    return 0;
}

/**
 * 在当前已连接的套接字上发送消息
 */
int pub_on_con_socket(zmq::socket_t *socket, zmsg &msg, string &identity, string &message)
{
    //设置消息
    msg.body_set(message.c_str());
    msg.wrap(identity.c_str(), NULL);
    msg.send(*socket);
    return 0;
}

/**
 * 发布订阅模式 发布
 */
int pub_zmq(string &message, string &url,string& topic)
{
    zmq::context_t context(1);
    zmq::socket_t publisher(context, ZMQ_PUB);
    publisher.bind(url.c_str());

    while (1) {
        s_sendmore (publisher, topic.c_str());
        s_send (publisher, message.c_str());
        Sleep (1);
    }
    return 0;
}

/**
 * 发布订阅模式 订阅
 */
int sub_zmq(string &url, SubMsgCallBack smcb,string& topic)
{
    zmq::context_t context(1);
    zmq::socket_t subscriber (context, ZMQ_SUB);
    subscriber.connect(url.c_str());
//    subscriber.setsockopt( ZMQ_SUBSCRIBE, "INSTANT", 1);
//    subscriber.setsockopt( ZMQ_SUBSCRIBE, "GROUP_APPRAISE", 1);
    subscriber.setsockopt( ZMQ_SUBSCRIBE, topic.c_str(), 1);
    while (1) {
          //  Read envelope with address
//        std::string address = s_recv (subscriber);
        //  Read message contents
        std::string contents = s_recv (subscriber);
        std::cout << contents << std::endl;
    }
    return 0;
}

int req_zmq(string &url, string &msg, ReqMsgCallBack cb)
{

    zmq::context_t context (1);

    zmq::socket_t * client = s_client_socket (context, url);
    int sequence = 0;
    int retries_left = REQUEST_RETRIES;
    while (retries_left) {
        std::stringstream request;
        request << ++sequence;
        s_send (*client, msg);
        Sleep (1);

        bool expect_reply = true;
        while (expect_reply) {
            //  Poll socket for a reply, with timeout
            zmq::pollitem_t items[] = {
                { static_cast<void*>(*client), 0, ZMQ_POLLIN, 0 } };
            zmq::poll (&items[0], 1, REQUEST_TIMEOUT);

            //  If we got a reply, process it
            if (items[0].revents & ZMQ_POLLIN) {
                //  We got a reply from the server, must match sequence
                std::string reply = s_recv (*client);
//                if (atoi (reply.c_str ()) == sequence) {
                    std::cout << "I: server replied OK (" << reply << ")" << std::endl;
                    expect_reply = false;
                    retries_left = 0;
//                }
//                else {
//                    std::cout << "E: malformed reply from server: " << reply << std::endl;
//                }
            }
            else
            if (--retries_left == 0) {
                std::cout << "E: server seems to be offline, abandoning" << std::endl;
                expect_reply = false;
                break;
            }
            else {
                std::cout << "W: no response from server, retrying…" << std::endl;
                //  Old socket will be confused; close it and open a new one
                delete client;
                client = s_client_socket (context, url);
                //  Send request again, on new socket
                s_send (*client, request.str());
            }
        }
    }
    delete client;


//    int major, minor, patch;
//    zmq_version (&major, &minor, &patch);
//    printf ("当前ZMQ版本号为 %d.%d.%d\n", major, minor, patch);

      // -----------------   原生C 代码  -----------------
//      void *context = zmq_ctx_new ();
//      void *requester = zmq_socket (context, ZMQ_REQ);
//      zmq_connect (requester, url.c_str());

      // 官网代码
//      int request_nbr;
//      for (request_nbr = 0; request_nbr != 1; request_nbr++) {
//          char buffer [10];
//          printf ("Sending Hello %d…\n", request_nbr);
//          zmq_send (requester, msg.c_str(), 10, 0);
////          zmq_recv (requester, buffer, 10, 0);
//          printf ("Received World %d\n", request_nbr);
//      }
      // 自己改的代码
//      int ret;
//      while (1)
//      {
//          zmq_send (requester, msg.c_str(), msg.size(), 0);
//          printf ("send msg 'hello----'\n");
//          sleep (1);
//          char buf[256];
//          ret = zmq_recv (requester, buf, sizeof(buf) - 1, 0);
//          buf[ret] = '\0';
//          printf ("recv msg %s\n", buf);
//          fflush(stdout);
//      }
//      // 正常是不会走到这一步的，以下只是演示我们应该如何结束
//      zmq_close (requester);
//      zmq_ctx_destroy (context);
      return 0;
}

int rep_zmq(string &url, RepMsgCallBack cb)
{
//    void *context = zmq_ctx_new();
//    //  与客户端通信的套接字
//    void *responder = zmq_socket (context, ZMQ_REP);
//    zmq_bind (responder, "tcp://*:5556");
//    int ret;
//    while (1)
//    {
//        //  等待客户端请求
//        char buf[256];
//        /*返回值代表读到的字节个数，失败返回-1并且设置errno*/
//        ret = zmq_recv (responder, buf, sizeof(buf) - 1, 0);
//        buf[ret] = '\0';
//        printf ("req -- > recv msg %s\n", buf);
//        sleep (5);
//        // 返回应答
//        memcpy (buf, "World", 5);
//        zmq_send (responder, buf, 5,  0);
//        fflush(stdout);
//    }
//    //  程序不会运行到这里，以下只是演示我们应该如何结束
//    zmq_close (responder);
//    zmq_ctx_destroy(context);

//    srandom ((unsigned) time (NULL));

    zmq::context_t context(1);
    zmq::socket_t server(context, ZMQ_REP);
    server.bind("tcp://*:5556");

    int cycles = 0;
    while (1) {
        std::string request = s_recv (server);
        cycles++;

//        // Simulate various problems, after a few cycles
//        if (cycles > 3 && within (3) == 0) {
//            std::cout << "I: simulating a crash" << std::endl;
//            break;
//        }
//        else
//        if (cycles > 3 && within (3) == 0) {
//            std::cout << "I: simulating CPU overload" << std::endl;
//            Sleep (2);
//        }
//        std::cout << "I: normal request (" << request << ")" << std::endl;
//        Sleep (1); // Do some heavy work
//        s_send (server, request);
        cout << "收到消息:" << request << endl;
        int elapsed = recordTime.msecsTo(QTime::currentTime());
        int t = duration - (elapsed/1000);
        QString st = QString("{\"code\":0,\"data\":{\"remainingTime\":%1}}").arg(t);
        s_send (server, st.toStdString());
    }
    return 0;
}

zmq::socket_t *s_client_socket(zmq::context_t &context,string& url)
{
    std::cout << "I: connecting to server…" << std::endl;
    zmq::socket_t * client = new zmq::socket_t (context, ZMQ_REQ);
    client->connect (url);

    //  Configure socket to not wait at close time
    int linger = 0;
    client->setsockopt (ZMQ_LINGER, &linger, sizeof (linger));
    return client;
}
